package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.KWSplit;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/1/6 08:55
 */
public class Dwd_01_DwsTrafficKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_01_DwsTrafficKeywordPageViewWindow().init(
            4001,
            2,
            "Dwd_01_DwsTrafficKeywordPageViewWindow"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //1. 建表与 topic 关联
        tEnv.executeSql("create table page_log(" +
                            " item string, " +
                            " last_page_id string, " +
                            " item_type string, " +
                            " ts bigint, " +
                            " et as to_timestamp_ltz(ts, 3), " +// timestamp(3)  时间属性
                            " watermark for et as et - interval '3' second " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "Dwd_01_DwsTrafficKeywordPageViewWindow"));
        
        // 2. 过滤出搜索记录, 从中取出搜索关键词
        Table itemTable = tEnv.sqlQuery("select " +
                                            " item," +
                                            " et " +
                                            "from page_log " +
                                            "where (last_page_id='home' or last_page_id='search') " +
                                            " and item_type='keyword' " +
                                            " and item is not null");
        tEnv.createTemporaryView("item_table", itemTable);
        
        // 3. 对关键词进行分词
        // 3.1 注册自定义的函数
        tEnv.createTemporaryFunction("kw_split", KWSplit.class);
        // 3.2 在 sql 中使用自定义函数
        Table kwTable = tEnv.sqlQuery("select " +
                                          " kw, " +
                                          " et " +  // 自定义函数中起的列名
                                          "from item_table " +
                                          "join lateral table( kw_split(item) ) on true");
        
        tEnv.createTemporaryView("kw_table", kwTable);
        // 4.开窗聚和
        // tvf
        Table result = tEnv.sqlQuery("select " +
                                         "date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                                         "date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                                         "kw keyword, " +
                                         "count(*) keyword_count," +
                                         "unix_timestamp() * 1000 as ts " +  // sum(1) count(*) count(id) count(1)
                                         "from table(tumble( table  kw_table, descriptor(et), interval '5' second ) ) " +
                                         "group by window_start, window_end, kw");
        // 5. 写出到 clickhouse 中
        /*
        1. 找到官方提供的 clickhouse connector
            目前没有提供
        2. 自定义 流的 sink
            把表转成流, 然后自定义 sink
            a: 完全自定义: 自己实现 jdbc 代码
                缺点: 性能差
            b: 在流的 jdbc 的基础上, 封装
            
         */
        
        // 5.1 把表转成流
        // false: 表示撤回  更新前, 或者删除
        // true: 新增或者更新后
        SingleOutputStreamOperator<KeywordBean> resultStream = tEnv
            .toRetractStream(result, KeywordBean.class)
            .filter(t -> t.f0)
            .map(t -> t.f1);
        // insert into user(a,b,c)values(?,?,?)
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_keyword_page_view_window", KeywordBean.class));
        
        
    
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    
    }
}
/*
dws层
	汇总层 聚和

	轻度聚合
		小窗口

		0-5 ..
		5-10 ...

	dwd(明细)-> dws(轻度聚和, 支持 sql, 支持实时. clickhouse. doris)-> ads(最终聚合)


	dwd(明细, 到 clickhouse) -> ads(直接对明细聚合)

搜索关键词粒度页面浏览各窗口汇总表
	
	0-5  手机  100
	0-5  电脑  200
	5-10 ..


  1. 找到用户的搜索记录, 从中取出搜索的关键词
  		"苹果 256g 白色手机"



  2. 按照关键词分组, 开窗聚合

  3. 结果写出到 clickhouse 中

 a: ddl 建表, 读取的 topic: 页面日志

 b: 过滤出搜索记录
    日志特点:
        last_page_id = home   && item_type=keyword  && item != null
        
        last_page_id = search && item_type=keyword  && item != null
    
 
 	"苹果256g白色手机"
 		苹果
 		256g
 		白色
 		手机
 	"华为512g黑色手机"
        华为
 		512g
 		黑色
 		手机

 	自定义函数:
 		1. scalar    一进一出	   udf
 		2. table     一进多出   udtf
 		3. aggregate 多进一出   udaf
 		4. TableAggregate 多进多出

 	中文分词器: IK 分词


 c: 开窗聚合
 	grouped window(过时)

 	tvf(推荐, 性能比较)
 		滚动
 		滑动
 		accumulate(累积)

 	over

 d: 写出
 */